package com.atguigu.utils;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;

import java.util.Properties;

public class MyKafkaUtil {

    private static Properties properties = new Properties();
    public static String KAFKA_SERVER = "hadoop102:9092,hadoop103:9092,hadoop104:9092";
    static {
        properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, KAFKA_SERVER);
    }

    public static FlinkKafkaConsumer<String> getKafkaSource(String topic, String groupId){
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);

        FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>(topic,
                new SimpleStringSchema(),
                properties);

        return consumer;
    }

    public static FlinkKafkaProducer<String> getKafkaSink(String topic){
//        properties.setProperty(ProducerConfig.ACKS_CONFIG, "-1");
        return new FlinkKafkaProducer<String>(topic,
                new SimpleStringSchema(),
                properties);
    }

    public static String getKafkaDDL(String sourceTopic, String groupId) {
        return "with( " +
                "    'connector' = 'kafka', " +
                "    'topic' = '" + sourceTopic + "', " +
                "    'properties.bootstrap.servers' = '" + KAFKA_SERVER + "', " +
                "    'properties.group.id' = '" + groupId + "', " +
                "    'format' = 'json', " +
                "    'scan.startup.mode' = 'latest-offset' " +
                ")";
    }

    public static String getUpsertKafkaDDL(String sinkTopic) {
        return "with( " +
                "    'connector' = 'upsert-kafka', " +
                "    'topic' = '" + sinkTopic + "', " +
                "    'properties.bootstrap.servers' = '" + KAFKA_SERVER + "', " +
                "    'key.format' = 'json', " +
                "    'value.format' = 'json' " +
                ")";
    }

    public static FlinkKafkaConsumer<String> getKafkaSource2(String topic, String groupId) {
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);

        return new FlinkKafkaConsumer<String>(topic,
                new KafkaDeserializationSchema<String>() {
                    @Override
                    public boolean isEndOfStream(String nextElement) {
                        return false;
                    }

                    @Override
                    public String deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
                        if (record.value() == null) {
                            return "null";
                        }
                        return new String(record.value());
                    }

                    @Override
                    public TypeInformation<String> getProducedType() {
                        return BasicTypeInfo.STRING_TYPE_INFO;
                    }
                },
                properties);
    }
}
